package com.borened.netty.protocol.mqtt;

import io.netty.channel.Channel;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.mqtt.*;
import io.netty.util.AttributeKey;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.annotation.Lazy;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

import java.util.List;

import static io.netty.channel.ChannelFutureListener.CLOSE_ON_FAILURE;
import static io.netty.handler.codec.mqtt.MqttQoS.AT_MOST_ONCE;


@Slf4j
@ChannelHandler.Sharable
@Component
public class MqttMessageHandler extends SimpleChannelInboundHandler<MqttMessage> {

    public static final AttributeKey<MqttClientSession> CLIENT_ATTRIBUTE_KEY = AttributeKey.valueOf("client");
    @Resource
    public SubscriberManager subscriberManager;

    @Resource
    public MqttPublisher mqttPublisher;

    @Resource
    @Lazy
    public MqttBroker broker;


    /**
     * <pre>
     *  <h2><a href="https://public.dhe.ibm.com/software/dw/webservices/ws-mqtt/mqtt-v3r1.html#retain-flag">MQTT V3协议文档</a></h2>
     *  <h2>MQTT 报文类型</h2>
     *
     *  <img src="https://img-blog.csdnimg.cn/b9435e712b6a420586f1a835b5b583c9.png"/>
     * </pre>
     *
     * @param ctx 信道上下文
     * @param mqttMessage the message to handle
     * @throws Exception
     */
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, MqttMessage mqttMessage) throws Exception {
        Channel channel = ctx.channel();
        MqttMessageType messageType = mqttMessage.fixedHeader().messageType();
        switch (messageType) {
            case CONNECT:
                handleConnect(channel,(MqttConnectMessage) mqttMessage);
                break;
            case DISCONNECT:
                handleDisconnect(channel);
                break;
            case PUBLISH:
                handlePublish(channel,(MqttPublishMessage) mqttMessage);
                break;
            case PUBACK:
                handlePubAck(channel,(MqttPubAckMessage) mqttMessage);
                break;

            //Qos2 发布收到
            case PUBREC:
                break;
            //Qos2 发布释放
            case PUBREL:
                break;
            //Qos2 发布完成
            case PUBCOMP:
                break;
            case SUBSCRIBE:
                handleSubscribe(channel,(MqttSubscribeMessage) mqttMessage);
                break;
            case UNSUBSCRIBE:
                handleUnsubscribe(channel,(MqttUnsubscribeMessage) mqttMessage);
                break;
            case PINGREQ:
                MqttFixedHeader pingHeader = new MqttFixedHeader(MqttMessageType.PINGRESP, false, AT_MOST_ONCE,
                        false, 0);
                MqttMessage pingResp = new MqttMessage(pingHeader);
                channel.writeAndFlush(pingResp).addListener(CLOSE_ON_FAILURE);
                break;
            default:
                break;
        }
    }

    private void handlePubAck(Channel channel, MqttPubAckMessage mqttMessage) {
        log.info("handlePubAck clientId:{},msgId:{}", channel.attr(CLIENT_ATTRIBUTE_KEY).get().getClientId(), mqttMessage.variableHeader().messageId());
    }

    private void handleDisconnect(Channel channel) {
        MqttClientSession mqttClientSession = channel.attr(CLIENT_ATTRIBUTE_KEY).get();
        String clientId = mqttClientSession.getClientId();
        log.info("handleDisconnect clientId:{}", clientId);
        channel.closeFuture().addListener(future -> {
            if (future.isSuccess()){
                broker.removeClient(mqttClientSession);
            }
        });
    }

    private void handleConnect(Channel channel, MqttConnectMessage message) {
        MqttConnectPayload payload = message.payload();
        String clientId = payload.clientIdentifier();
        boolean cleanSession = message.variableHeader().isCleanSession();
        int keepAlive = message.variableHeader().keepAliveTimeSeconds();
        log.info("handleConnect clientId:{} cleanSession:{} keepAlive:{}", clientId, cleanSession, keepAlive);
        //todo mqtt AUTH
        MqttClientSession client = new MqttClientSession(clientId, cleanSession, keepAlive,channel);
        broker.addClient(client);
        //ack
        MqttConnAckMessage connAckMessage = MqttMessageBuilders.connAck().returnCode(MqttConnectReturnCode.CONNECTION_ACCEPTED).build();
        channel.writeAndFlush(connAckMessage).addListener(future ->{
            if (!future.isSuccess()){
                channel.close();
                broker.removeClient(client);
                log.error("connect ack error:{}", future.cause().getMessage());
            }else {
                channel.attr(CLIENT_ATTRIBUTE_KEY).set(client);
            }
        });
    }

    private void handlePublish(Channel channel, MqttPublishMessage mqttMessage) {
        // handle PUBLISH message
        String topic = mqttMessage.variableHeader().topicName();
        mqttPublisher.publicQos0(topic,mqttMessage);

        //qos1 pub ack

        //qos2
    }

    private void handleSubscribe(Channel channel, MqttSubscribeMessage mqttMessage) {
        // handle SUBSCRIBE message
        List<MqttTopicSubscription> mqttTopicSubscriptions = mqttMessage.payload().topicSubscriptions();
        mqttTopicSubscriptions.forEach(mqttTopicSubscription -> {
            String topic = mqttTopicSubscription.topicName();
            int qos = mqttTopicSubscription.qualityOfService().value();
            log.info("handleSubscribe topic:{} qos:{}", topic, qos);
            subscriberManager.addSubscriber(new Subscriber(topic,channel.attr(CLIENT_ATTRIBUTE_KEY).get()));
        });
        //sub ack
        MqttSubAckMessage subAckMessage = MqttMessageBuilders.subAck().packetId(mqttMessage.idAndPropertiesVariableHeader().messageId())
                .addGrantedQos(mqttMessage.fixedHeader().qosLevel()).build();
        channel.writeAndFlush(subAckMessage).addListener(future ->{
            if (!future.isSuccess()){
                log.error("subscribe ack error:{}", future.cause().getMessage());
            }
        });
    }

    private void handleUnsubscribe(Channel channel, MqttUnsubscribeMessage mqttMessage) {
        // handle UNSUBSCRIBE message
        List<String> unsubscribeTopics = mqttMessage.payload().topics();
        unsubscribeTopics.forEach(topic -> {
            subscriberManager.removeSubscriber(new Subscriber(topic,channel.attr(CLIENT_ATTRIBUTE_KEY).get()));
        });
        MqttUnsubAckMessage subAckMessage = MqttMessageBuilders.unsubAck().packetId(mqttMessage.idAndPropertiesVariableHeader().messageId())
                .build();
        channel.writeAndFlush(subAckMessage);
    }
}